package com.gmail.dengtao.joe.group.channel.impl;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.gmail.dengtao.joe.commons.util.StringUtils;
import com.gmail.dengtao.joe.group.exception.ChannelException;
import com.gmail.dengtao.joe.group.exception.MemberInvalidException;
import com.gmail.dengtao.joe.group.exception.MemberNotReachableException;
import com.gmail.dengtao.joe.group.interceptor.InterceptorStack;
import com.gmail.dengtao.joe.group.member.Member;
import com.gmail.dengtao.joe.group.member.MemberUtils;
import com.gmail.dengtao.joe.group.transport.ChannelReceiver;
import com.gmail.dengtao.joe.group.transport.ChannelSender;
import com.gmail.dengtao.joe.group.transport.impl.TCPChannelReceiver;
import com.gmail.dengtao.joe.group.transport.impl.TCPChannelSender;

public class TCPChannel extends AbstractChannel {

	private static final Logger LOGGER = LoggerFactory.getLogger(TCPChannel.class);
	
	private final Object receiverLock = new Object();
	private final Object senderLock = new Object();
	
	private TCPChannel channel = this;
	
	private Thread channelSenderThread;
	private Thread channelReceiverThread;
	
	private Map<String, Member> memberMap = new ConcurrentHashMap<String, Member>();
	private Member local;

	public TCPChannel(String host, int port) throws ChannelException {
		
		local = new Member(host, port);
		
		// Logger...
		try {
			channelReceiver = new TCPChannelReceiver(host, port, this);
		} catch (Exception e) {
			close();
			throw new ChannelException("ChannelReceiver initialize failed", e);
		}
		try {
			channelSender = new TCPChannelSender(host, port, this);
		} catch (Exception e) {
			close();
			throw new ChannelException("ChannelSender failed to initialize", e);
		}
		interceptorStack = new InterceptorStack();
		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("[CHANNEL][INITIALIZE][SUCCESS]");
		}
	}

	@Override
	public synchronized void open() {
		if (isOpen()) {
			throw new IllegalStateException("Channel is already started!");
		} else {
			setOpen(true);
			// Get channel id.
			ChannelFactory.openChannel(channel);
			if (LOGGER.isDebugEnabled()) {
				LOGGER.debug("[CHANNEL][OPENING...]");
			}
		}
		if (membershipListener == null) {
			throw new IllegalStateException("The MembershipListener for this Channel was not set! call 'setMembershipListener(MembershipListener membershipListener)' to set MembershipListener for this Channel.");
		} else {
			interceptorStack.setMembershipListener(membershipListener);
		}
		if (channelListener == null) {
			throw new IllegalStateException("The ChannelListener for this Channel was not set! call 'setChannelListener(ChannelListener channelListener)' to set ChannelListener for this Channel.");
		} else {
			interceptorStack.setChannelListener(channelListener);
		}
		
		try {
			synchronized (receiverLock) {
				// prepare receiver
				channelReceiver.setInterceptorStack(interceptorStack);
				channelReceiverThread = new ChannelReceiverThread(channelReceiver, receiverLock);
				channelReceiverThread.setDaemon(true);
				channelReceiverThread.setName("ChannelReceiverThread");
				// start reveiver
				channelReceiverThread.start();
				
				receiverLock.wait();
				
				synchronized (senderLock) {
					// prepare sender
					channelSender.setInterceptorStack(interceptorStack);
					channelSenderThread = new ChannelSenderThread(channelSender, senderLock);
					channelSenderThread.setDaemon(true);
					channelSenderThread.setName("ChannelSenderThread");
					// start sender
					channelSenderThread.start();
					
					senderLock.wait();
				}
			}
		} catch (InterruptedException e) {
			if (LOGGER.isDebugEnabled()) {
				LOGGER.error("[CHANNEL][EXCEPTION][OPEN FAILED]", e);
			}
		}
	}

	@Override
	public synchronized void close() {
		if (!isClose()) {
			setOpen(false);
			if (LOGGER.isDebugEnabled()) {
				LOGGER.debug("[CHANNEL][CLOSING...]");
			}
			try {
				channelSender.stop();
			} catch (Exception e) {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.error("[CHANNEL][SENDER][STOP FAILED]", e);
				}
			}
			try {
				channelReceiver.stop();
			} catch (Exception e) {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.error("[CHANNEL][RECEIVER][STOP FAILED]", e);
				}
			}
			if (LOGGER.isDebugEnabled()) {
				LOGGER.debug("[CHANNEL][CLOSED]");
			}
			setClose(true);
			
			// release channel id.
			ChannelFactory.closeChannel(channel);
		}
	}
	
	@Override
	public void addMember(Member member) {
		if (StringUtils.isBlank(member.getLocalUuid())) {
			member.setLocalUuid(MemberUtils.getUniqueId(local, member));
		}
		if (!memberMap.containsKey(member.getLocalUuid())) {
			try {
				memberMap.put(member.getLocalUuid(), member);
				channelSender.connect(member);
			} catch (Exception e) {
				memberMap.remove(member.getLocalUuid());
				interceptorStack.fireExceptionCaught(this, member, new MemberNotReachableException(e));
			}
		} else {
			if (member.hasStatus(Member.Status.RECEIVER_NOT_CONNECT)) {
				try {
					channelSender.connect(member);
				} catch (Exception e) {
					interceptorStack.fireExceptionCaught(this, member, new MemberNotReachableException(e));
				}
			}
		}
	}

	@Override
	public void removeMember(Member member) {
		if (StringUtils.isBlank(member.getLocalUuid())) {
			member.setLocalUuid(MemberUtils.getUniqueId(local, member));
		}
		if (memberMap.containsKey(member.getLocalUuid())) {
			member = memberMap.get(member.getLocalUuid());
			// Close sender...
			if (member.getSender() != null) member.getSender().close();
			// Close receiver...
			if (member.getReceiver() != null) member.getReceiver().close();
			memberMap.remove(member.getLocalUuid());
		}
	}
	
	@Override
	public Member[] getMembers() {
		return memberMap.values().toArray(new Member[memberMap.values().size()]);
	}

	@Override
	public void send(Member target, Object data) {
		if (target == null) {
			throw new MemberInvalidException();
		}
		if (data == null) {
			throw new IllegalArgumentException();
		}
		if (!target.hasStatus(Member.Status.AVAILABLE)) {
			throw new MemberNotReachableException();
		}
		channelSender.send(target, data);
	}
	
	@Override
	public void send(Member[] destination, Object data) {
		for (Member target : destination) {
			if (target.hasStatus(Member.Status.AVAILABLE)) {
				send(target, data);
			} else {
				try {
					interceptorStack.fireExceptionCaught(channel, target, new MemberNotReachableException());
					interceptorStack.fireDataNotSent(channel, target, data);
				} catch (Exception e) {
					interceptorStack.fireExceptionCaught(channel, target, new ChannelException(e));
				}
			}
		}
	}

	@Override
	public void broadcast(Object data) {
		Iterator<Member> iter = memberMap.values().iterator(); 
		for ( ; iter.hasNext(); ) {
			Member target = iter.next();
			if (target.hasStatus(Member.Status.AVAILABLE)) {
				send(target, data);
			} else {
				try {
					interceptorStack.fireExceptionCaught(channel, target, new MemberNotReachableException());
					interceptorStack.fireDataNotSent(channel, target, data);
				} catch (Exception e) {
					interceptorStack.fireExceptionCaught(channel, target, new ChannelException(e));
				}
			}
		}
	}
	
	/**
	 * Channel Reveiver Thread
	 * @author <a href="mailto:joe.dengtao@gmail.com">DengTao</a>
	 * @version 1.0
	 * @since 1.0
	 */
	class ChannelReceiverThread extends Thread {

		private ChannelReceiver channelReceiver;	
		private final Object receiverLock;
	
		public ChannelReceiverThread(ChannelReceiver channelReceiver, Object receiverLock) {
			this.channelReceiver = channelReceiver;
			this.receiverLock = receiverLock;
		}

		@Override
		public void run() {
			try {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[CHANNEL][RECEIVER][STARTED][LISTEN /" + channelReceiver.getHost() + ":" + channelReceiver.getPort() + "]");
				}
				channelReceiver.start(receiverLock);
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[CHANNEL][RECEIVER][STOPPED...]");
				}
			} catch (Exception e) {
				if (channel.isOpen()) {
					if (LOGGER.isDebugEnabled()) {
						LOGGER.error("[CHANNEL][RECEIVER][START FAILED]", e);
					}
					channel.close();
				}
			}
		}
		
	}
	
	/**
	 * Channel Sender Thread
	 * @author <a href="mailto:joe.dengtao@gmail.com">DengTao</a>
	 * @version 1.0
	 * @since 1.0
	 */
	class ChannelSenderThread extends Thread {
		
		private ChannelSender channelSender;
		private final Object senderLock;
		
		public ChannelSenderThread(ChannelSender channelSender, Object senderLock) {
			this.channelSender = channelSender;
			this.senderLock = senderLock;
		}

		@Override
		public void run() {
			try {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[CHANNEL][SENDER][STARTED]");
				}
				channelSender.start(senderLock);
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[CHANNEL][SENDER][STOPPED]");
				}
			} catch (Exception e) {
				if (channel.isOpen()) {
					if (LOGGER.isDebugEnabled()) {
						LOGGER.error("[CHANNEL][SENDER][START FAILED]", e);
					}
					channel.close();
				}
			}
		}
		
	}

	@Override
	public String toString() {
		return "TCPChannel [id=" + id + ", open=" + open + ", close=" + close + "]";
	}

	@Override
	public Member getMember(String uuid) {
		if (StringUtils.isBlank(uuid)) {
			return null;
		} else {
			Member member = memberMap.get(uuid);
			if (member != null) {
			}
			return member;
		}
	}
	
}